/**
 * Description:
 * Class Name:PubSubTest
 * Date:2022/5/17 16:19
 *
 * @author LTisme
 * @version JAVA17
 */
package 疯狂Java讲义.第16章_多线程._16_9_线程相关类._16_9_4_Java9新增的发布_订阅框架;

import java.util.List;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

class MySubscriber<T> implements Flow.Subscriber<T>{
    // 发布者与订阅者之间的纽带
    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        // 开始请求数据
        subscription.request(1);
    }

    @Override
    public void onNext(T item) {
        System.out.println("获取到的数据是：" + item);
        // 请求下一条数据
        subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        throwable.printStackTrace();
        synchronized ("fkjava"){
            "fkjava".notifyAll();
        }
    }

    @Override   // 订阅结束时自动触发该方法
    public void onComplete() {
        System.out.println("订阅结束");
        synchronized ("fkjava"){
            "fkjava".notifyAll();
        }
    }
}

public class PubSubTest {
    public static void main(String[] args) {
        // 创建一个SubmissionPublisher 作为发布者
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        // 创建订阅者
        MySubscriber<String> subscriber = new MySubscriber<>();
        // 注册订阅者
        publisher.subscribe(subscriber);
        // 发布几个数据项
        System.out.println("开发者发布数据……");
        List.of("Java", "Kotlin", "Go", "Erlang", "Swift", "Lua").forEach(im -> {
            // 提交数据
            publisher.submit(im);
            try{
                Thread.sleep(500);
            }
            catch (Exception e){
                e.printStackTrace();
            }
        });
    }
}
